import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

public class StudentDataProducer {

    private static final String KAFKA_BROKERS = "192.168.40.43:9092";
    private static final String TOPIC_NAME = "term6Program";
    private static final long SLEEP_INTERVAL_MS = 5000;

    private static final String[] CLASS_IDS = {"210715", "210716", "220715", "220716"};

    private static final String[] LAST_NAMES = {"王", "李", "张", "刘", "陈", "杨", "黄", "赵", "吴", "周", "徐", "孙", "马", "朱", "胡", "郭"};

    private static final String[] FIRST_NAMES_MALE = {
            "伟", "强", "军", "杰", "明", "超", "浩", "宇", "轩", "航", "雨", "俊", "嘉", "博", "梓", "子"
    };

    private static final String[] FIRST_NAMES_FEMALE = {
            "敏", "静", "丽", "娜", "芳", "婷", "雨", "欣", "梓", "诗", "梦", "秀", "桂", "淑", "玉", "红"
    };

    private static final String[] FIRST_NAMES_TWOCHAR_MALE = {
            "浩然", "宇轩", "子轩", "宇航", "雨浩", "俊杰", "嘉豪", "博文", "梓晨", "子墨"
    };

    private static final String[] FIRST_NAMES_TWOCHAR_FEMALE = {
            "雨桐", "欣怡", "梓涵", "诗涵", "梦琪", "秀英", "桂英", "淑芬", "玉兰", "红梅"
    };

    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd");
    private static long studentIdCounter = 2022L;

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<>(props);
        Random random = new Random();

        System.out.println("开始向 Kafka 主题 '" + TOPIC_NAME + "' 推送学生数据...");
        System.out.println("每 " + SLEEP_INTERVAL_MS + " 毫秒推送一条数据。按 Ctrl+C 停止。");

        try {
            while (true) {
                String classId = CLASS_IDS[random.nextInt(CLASS_IDS.length)];

                String name;
                int gender = random.nextInt(2);

                if (gender == 1) {
                    String lastName = LAST_NAMES[random.nextInt(LAST_NAMES.length)];
                    if (random.nextBoolean()) {
                        String firstName = FIRST_NAMES_MALE[random.nextInt(FIRST_NAMES_MALE.length)];
                        name = lastName + firstName;
                    } else {
                        String firstName = FIRST_NAMES_TWOCHAR_MALE[random.nextInt(FIRST_NAMES_TWOCHAR_MALE.length)];
                        name = lastName + firstName;
                    }
                } else {
                    String lastName = LAST_NAMES[random.nextInt(LAST_NAMES.length)];
                    if (random.nextBoolean()) {
                        String firstName = FIRST_NAMES_FEMALE[random.nextInt(FIRST_NAMES_FEMALE.length)];
                        name = lastName + firstName;
                    } else {
                        String firstName = FIRST_NAMES_TWOCHAR_FEMALE[random.nextInt(FIRST_NAMES_TWOCHAR_FEMALE.length)];
                        name = lastName + firstName;
                    }
                }

                String dob = generateRandomDOB(1998, 2006);
                String studentId = String.valueOf(studentIdCounter++);
                if (studentIdCounter > 99999999) {
                    studentIdCounter = 20220000L;
                }

                int semester = random.nextInt(8) + 1;
                double gpa = 1.0 + random.nextDouble() * 9.0;
                gpa = Math.min(gpa, 10.0);
                gpa = Math.max(gpa, 1.0);
                String status = (semester == 8 && random.nextDouble() > 0.3) ? "L" : "A";

                String recordValue = String.join("\t", classId, name, String.valueOf(gender), dob, studentId,
                        String.valueOf(semester), String.format("%.2f", gpa), status);

                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, studentId, recordValue);
                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        System.out.printf("发送成功: 分区=%d, 偏移量=%d, 姓名=%s, 班级=%s%n",
                                metadata.partition(), metadata.offset(), name, classId);
                    } else {
                        System.err.printf("发送失败: 姓名=%s, 错误: %s%n", name, exception.getMessage());
                    }
                });

                TimeUnit.MILLISECONDS.sleep(SLEEP_INTERVAL_MS);
            }
        } catch (InterruptedException e) {
            System.err.println("生产者线程被中断。");
            Thread.currentThread().interrupt();
        } finally {
            producer.flush();
            producer.close();
        }
    }

    private static String generateRandomDOB(int startYear, int endYear) {
        Calendar calendar = Calendar.getInstance();
        int year = ThreadLocalRandom.current().nextInt(startYear, endYear + 1);
        int month = ThreadLocalRandom.current().nextInt(0, 12);
        calendar.set(year, month, 1);
        int dayOfMonth = ThreadLocalRandom.current().nextInt(1, calendar.getActualMaximum(Calendar.DAY_OF_MONTH) + 1);
        calendar.set(Calendar.DAY_OF_MONTH, dayOfMonth);
        return DATE_FORMAT.format(calendar.getTime());
    }
}